package com.atguigu.actual.edu0417.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.actual.edu0417.app.BaseApp;

import com.atguigu.actual.edu0417.beans.TradeProvinceOrderBean;
import com.atguigu.actual.edu0417.func.DimAsyncFunction;
import com.atguigu.actual.edu0417.unils.DateFormatUtil;
import com.atguigu.actual.edu0417.unils.DorisSinkUtil;
import com.atguigu.actual.edu0417.unils.MyKafkaUtil;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;



import java.math.BigDecimal;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;


/**
 * @author Mr.Ren
 * @create 2023-10-21 16:46
 * 来源和省份粒度交易聚合统计
 */
public class DwsTradeProvinceOrderWindow extends BaseApp {
    public static void main(String[] args) {
        DwsTradeProvinceOrderWindow dwsTradeProvinceOrderWindow = new DwsTradeProvinceOrderWindow();
        dwsTradeProvinceOrderWindow.base(args);
    }

    @Override
    public void process(StreamExecutionEnvironment env) {
        //TODO 1 从kafka主题中读取数据
        //声明消费主题和消费者组
        String topic = "dwd_trade_order_detail";
        String groupId= "dws_trade_province_order_group";
        //创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //消费数据 分装为流
        DataStreamSource<String> kafkaStrDS
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        //过滤空消息,第流中数据类型进行转换 jsonStr -> jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            out.collect(jsonObj);
                        }
                    }
                }
        );

        //jsonObjDS.print(">>");
        //第流中数据类型进行转换 jsonObj -> 实体类对象
        SingleOutputStreamOperator<TradeProvinceOrderBean> orderBeanDS = jsonObjDS.map(
                new MapFunction<JSONObject, TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean map(JSONObject jsonObj) throws Exception {
                        String provinceId = jsonObj.getString("province_id");
                        String sourceId = jsonObj.getString("source_id");
                        String originAmount = jsonObj.getString("origin_amount");
                        String couponReduce = jsonObj.getString("coupon_reduce");
                        String finalAmount = jsonObj.getString("final_amount");
                        Long ts = jsonObj.getLong("ts") * 1000;
                        String orderId = jsonObj.getString("order_id");
                        String userId = jsonObj.getString("user_id");

                        TradeProvinceOrderBean orderBean = TradeProvinceOrderBean.builder()
                                .provinceId(provinceId)
                                .sourceId(sourceId)
                                .orderOriginAmount(new BigDecimal(originAmount))
                                .orderCouponReduceAmount(new BigDecimal(couponReduce))
                                .orderFinalAmount(new BigDecimal(finalAmount))
                                .orderIdSet(new HashSet(Collections.singleton(orderId)))
                                .userIdSet(new HashSet(Collections.singleton(userId)))
                                .ts(ts)
                                .build();
                        return orderBean;
                    }
                }
        );
        //orderBeanDS.print(">>>");
        //指定watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeProvinceOrderBean> withWatermarkDS = orderBeanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradeProvinceOrderBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TradeProvinceOrderBean>() {
                                    @Override
                                    public long extractTimestamp(TradeProvinceOrderBean orderBean, long recordTimestamp) {
                                        return orderBean.getTs();
                                    }
                                }
                        )
        );
        //按照统计的维度(province_id,source_id)进行分组
        KeyedStream<TradeProvinceOrderBean, Tuple2<String, String>> keyedBeanDS = withWatermarkDS.keyBy(
                new KeySelector<TradeProvinceOrderBean, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> getKey(TradeProvinceOrderBean bean) throws Exception {
                        return Tuple2.of(
                                bean.getProvinceId(),
                                bean.getSourceId()
                        );
                    }
                }
        );
        //开窗
        WindowedStream<TradeProvinceOrderBean, Tuple2<String, String>, TimeWindow> windowDS
                = keyedBeanDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        //聚合计算
        SingleOutputStreamOperator<TradeProvinceOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeProvinceOrderBean>() {
                    @Override
                    public TradeProvinceOrderBean reduce(TradeProvinceOrderBean value1, TradeProvinceOrderBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        value1.getUserIdSet().addAll(value2.getUserIdSet());
                        value1.setOrderFinalAmount(value1.getOrderFinalAmount().add(value2.getOrderFinalAmount()));
                        return value1;
                    }
                },
                new ProcessWindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, Tuple2<String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple2<String, String> stringStringTuple2, ProcessWindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, Tuple2<String, String>, TimeWindow>.Context context, Iterable<TradeProvinceOrderBean> elements, Collector<TradeProvinceOrderBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        String cur_date = DateFormatUtil.toDate(context.window().getStart());
                        String Date = DateFormatUtil.toDate(context.window().getStart());
                        for (TradeProvinceOrderBean orderBean : elements) {
                            orderBean.setStt(stt);
                            orderBean.setEdt(edt);
                            orderBean.setCurDate(cur_date);
                            orderBean.setOrderCount((long) orderBean.getOrderIdSet().size());
                            orderBean.setOrderNumber((long) orderBean.getUserIdSet().size());
                            out.collect(orderBean);

                        }
                    }
                }
        );
        //reduceDS.print(">>>");
        //关联省份维度
        SingleOutputStreamOperator<TradeProvinceOrderBean> withProvinceDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TradeProvinceOrderBean>("dim_base_province") {
                    @Override
                    public void join(TradeProvinceOrderBean orderBean, JSONObject dimCourseJsonObj) {
                        orderBean.setProvinceName(dimCourseJsonObj.getString("name"));
                    }

                    @Override
                    public String getKey(TradeProvinceOrderBean orderBean) {
                        return orderBean.getProvinceId();
                    }
                },
                60, TimeUnit.SECONDS
        );
        //关联来源维度
        SingleOutputStreamOperator<TradeProvinceOrderBean> withSourceDS = AsyncDataStream.unorderedWait(
                withProvinceDS,
                new DimAsyncFunction<TradeProvinceOrderBean>("dim_base_source") {
                    @Override
                    public void join(TradeProvinceOrderBean orderBean, JSONObject dimCourseJsonObj) {
                        orderBean.setSourceName(dimCourseJsonObj.getString("source_site"));
                    }

                    @Override
                    public String getKey(TradeProvinceOrderBean orderBean) {
                        return orderBean.getSourceId();
                    }
                },
                60, TimeUnit.SECONDS
        );
        withSourceDS.print(">>>>");
        //将关联的结果写到Doris中
        withSourceDS.addSink(DorisSinkUtil.getJdbcSink("insert into dws_trade_province_order_window values (?,?,?,?,?,?,?,?,?,?,?,?)"));


    }
}
